package edu.cornell.cs.lsi.mapreduce.pass3.test;





import static edu.cornell.cs.lsi.mapreduce.Constants.G;
import static edu.cornell.cs.lsi.mapreduce.Constants.Lmax_SQUARED;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.Map.Entry;

import edu.cornell.cs.lsi.mapreduce.Constants;
import edu.cornell.cs.lsi.mapreduce.pass1.DoublePoint;



public class FasterTester {
		
	public static final double Lmax_SQUARED = Constants.Lmax * Constants.Lmax;
	

	
	public static void main(String[] args) throws Exception{
		
		String inputDataPrefix = "C:\\cygwin\\bin\\hadoop-0.20.2\\bin\\";
		String inputCalcPrefix = "C:\\cygwin\\bin\\hadoop-0.20.2\\bin\\outdir3\\";
		
		String inputDataFileName = "smallprod.txt";
		String inputCalcFileName = "part-r-00000";
		
		String outputPath = "C:\\cygwin\\out.txt";
		
		FileWriter writer = new FileWriter(outputPath);
		
		
		String inputDataPath = inputDataPrefix + inputDataFileName;
		String inputCalcPath = inputCalcPrefix + inputCalcFileName;
		
		
		Timer t = new Timer();
				
		t.reset("load points from file");
		List<DPoint> allPoints = getPointsFromDataFiles(inputDataPath);
		
		t.markAndReset("sort allPoints");
		
		Collections.sort(allPoints);
		
		t.mark();
		
		int edgeCount = 0;
		//edgeCount = getCorEdgeCount(allPoints);
	
		List<Link> calcLinks = null;
		
		calcLinks = getCalcLinks(inputCalcPath);
		
		//printLinkList(calcLinks);
		
		System.out.println("\n\n------------\n");
		
		t.reset("getCorLinks");
		
		List<Link> corLinks = getCorLinks(allPoints);
		
		t.mark();
		
		//printLinkList(corLinks);
		
		compareLinks(calcLinks, corLinks, edgeCount);
		
		//System.out.println(edgeCount);
		
		/*
		for(Link l : corLinks){
			writer.append(l.toString() + "\n");
		}
		writer.flush();
		writer.close();*/
		
	}

	private static void printLinkList(List<Link> links){
		for(Link l : links){
			System.out.println(l);
		}
	}
	
	
	private static void compareLinks(List<Link> calcLinks, List<Link> corLinks, int corEdgeCount){
		
		int i;
		Link curCalcLink;
		Link curCorLink;
		
		
		if(calcLinks.size() != corLinks.size()){
			System.out.println("Failure: wrong size.");
			System.out.println(String.format("Expected: %d, found: %d", corLinks.size(), calcLinks.size()));
			System.out.println("------ colinks ");
			
			for(Link link : corLinks){
				//System.out.println(link.child.x+" "+link.child.y+" --> "+link.parent.x+" "+link.parent.y);
			}
			return;
		}
		
		for(i = 0; i < calcLinks.size(); i++){
			curCalcLink = calcLinks.get(i);
			curCorLink = corLinks.get(i);
			if(!curCalcLink.equals(curCorLink)){
				System.out.println("Failure: unequal");
				System.out.println("Expected: " + curCorLink.toString());
				System.out.println("Found: " + curCalcLink.toString());
				
				return;
			}
		}
		/*
		for(Link l : calcLinks){
			if(l.edgeCount > -1){
				calcEdgeCount += l.edgeCount;
			}
		}
		
		if(calcEdgeCount != corEdgeCount){
			System.out.println(String.format("Edge count mismatch. Expected %d, got %d.", corEdgeCount, calcEdgeCount));
			return;
		}*/
		
		System.out.println("Pass");
	}
	
	
	private static List<Link> getCorLinks(List<DPoint> allPoints){
		List<Link> corLinks = new ArrayList<Link>();
		SimpleUnion u;
		
		int i;
		int j;
		
		Timer t = new Timer();
		
        Comparator<DPoint> yComp = new Comparator<DPoint>() {

            @Override
            public int compare(DPoint arg0, DPoint arg1) {
                if (arg0.y < arg1.y)
                    return -1;
                if (arg0.y > arg1.y)
                    return 1;
                if (arg0.x < arg1.x)
                    return -1;
                if (arg0.x > arg1.x)
                    return 1;
                return 0;
            }

        };
		
        TreeSet<DPoint> tree = new TreeSet<DPoint>(yComp);
        
        DPoint minPoint;
        DPoint maxPoint;
       
        
		t.reset("build union");
		
		u = new SimpleUnion();
		for(DPoint f : allPoints){
			u.add(f);
		}
			
		t.markAndReset("run for loop");
		
		
		int head;
		int tail;
		
		head = 0;
		tail = 0;
		DPoint headPoint;
		SortedSet<DPoint> pointsInSpan;		
		
		
		for(head = 0; head < allPoints.size(); head++){

			headPoint = allPoints.get(head);
			
	        minPoint = new DPoint(-1d, headPoint.y - Constants.Lmax);
	        maxPoint = new DPoint(2f, headPoint.y + Constants.Lmax);
	        pointsInSpan = tree.subSet(minPoint, maxPoint);

            for (DPoint pointInSpan : pointsInSpan) {

                // System.out.println("+++"+ pointInSpan.toString());

                if (pointInSpan.withinRange(headPoint, Lmax_SQUARED)) {
                    // if we're keeping track of # of edges, this is where we would do something

                    u.union(pointInSpan, headPoint);
                }
            }

            if (head + 1 < allPoints.size()) {
                while (tail < head && allPoints.get(tail).x < headPoint.x - Constants.Lmax) {
                    tree.remove(allPoints.get(tail));
                    tail++;
                }
            }
            tree.add(headPoint);
			
			if(head % 10000 == 0){
				System.out.println("head="+head+",   tail=" + tail);
			}
			
		}
		
		
		t.markAndReset("build link set");
		
		for(Entry<DPoint, DPoint> e : u.parentMap.entrySet()){
			corLinks.add(new Link(e.getKey(), e.getValue()));
		}
			
		
        System.out.println("Correct Num Discrete Components: " + u.childMap.size());
        int totalSize = 0;
        for (List<DPoint> l : u.childMap.values()) {
            totalSize += l.size() * l.size();
        }
        System.out.println("Correct Expected Size: " + (double) totalSize / u.parentMap.size());
		
		
		Collections.sort(corLinks);
		return corLinks;
	}

	
	private static int getCorEdgeCount(List<DPoint> allPoints){
		int edgeCount = 0;
		int i;
		int j;
		
		for(i = 1; i < allPoints.size(); i++){
			for(j = 0; j < i; j++){
				if(allPoints.get(i).withinRange(allPoints.get(j), Lmax_SQUARED)){
					edgeCount++;
				}
			}
		}
		return edgeCount;
	}

	private static List<Link> getCalcLinks(String path) throws Exception{
		List<Link> calcLinks = new ArrayList<Link>();
		
		BufferedReader dataReader = new BufferedReader(new FileReader(path));

		
		String line = dataReader.readLine();
		String[] majorSplit;
		String[] minorSplit;
		double x;
		double y;
		
		Link l;
		
		while(line != null){
			majorSplit = line.split("\t");
			minorSplit = majorSplit[0].split(" ");

			l = new Link();
			
			x = Double.parseDouble(minorSplit[0]);
			y = Double.parseDouble(minorSplit[1]);

			l.child = new DPoint(x,y);

			minorSplit = majorSplit[1].split(" ");
			
			x = Double.parseDouble(minorSplit[0]);
			y = Double.parseDouble(minorSplit[1]);

			l.parent = new DPoint(x,y);
			
			calcLinks.add(l);
			line = dataReader.readLine();
		}
		
		Collections.sort(calcLinks);
		
		return calcLinks;
	}
	
	private static List<DPoint> getPointsFromDataFiles(String path) throws Exception{
		List<DPoint> allPoints = new ArrayList<DPoint>();
		
		BufferedReader dataReader = new BufferedReader(new FileReader(path));

		
		String line = dataReader.readLine();
		String[] split;
		double x;
		double y;
		double weight;
		
		while(line != null){
			split = line.split(" ");
			x = Double.parseDouble(split[0]);
			y = Double.parseDouble(split[1]);
			weight = Double.parseDouble(split[2]);
			
			if(Constants.MIN_WEIGHT <= weight && Constants.MAX_WEIGHT > weight){
				allPoints.add(new DPoint(x, y));
			}
			
			line = dataReader.readLine();
		}
		
		return allPoints;
	}
	
}


class Timer{
	long startTime;
	String label;
	
	Timer(){
		
	}
	
	void reset(String label){
		this.label = label;
		startTime = System.currentTimeMillis();
	}
	
	void mark(){
		long end = System.currentTimeMillis();
		System.out.println(label + " took " + (end - startTime) + " millis");
	}
	
	void markAndReset(String newLabel){
		long end = System.currentTimeMillis();
		System.out.println(label + " took " + (end - startTime) + " millis");
		label = newLabel;
		startTime = System.currentTimeMillis();
	}
	
}